package demo.spark.hurypoint.logpaser

import java.util

import demo.spark.hurypoint.logpaser.SeqAccumulator.Trans
import org.apache.spark.util.AccumulatorV2

class SeqAccumulator[IN <: java.io.Serializable, OUT_ITEM <: java.io.Serializable](transF: IN => OUT_ITEM)
    extends AccumulatorV2[IN, java.util.List[OUT_ITEM]]
    with Trans[IN,OUT_ITEM] {

  private val _seq: java.util.List[OUT_ITEM] = new java.util.ArrayList;

  override def trans(origin: IN): OUT_ITEM = transF.apply(origin);

  override def isZero: Boolean = {
    _seq.size() == 0
  }

  override def copy(): AccumulatorV2[IN, util.List[OUT_ITEM]] = {
    val copy = new SeqAccumulator[IN, OUT_ITEM](transF)
    for (idx <- 1 until _seq.size()){
      copy._seq.add(_seq.get(idx))
    }
    copy
  }

  override def reset(): Unit = _seq.clear()
  override def add(v: IN): Unit = _seq.add(trans(v))
  override def merge(other: AccumulatorV2[IN, util.List[OUT_ITEM]]): Unit = _seq.addAll(other.value)
  override def value: util.List[OUT_ITEM] = _seq
}

object SeqAccumulator{
  trait Trans[IN, +Res]{
    def trans(origin: IN): Res;
  }
}
